package com.stoneframe.strokeserver.service;

import java.util.ArrayList;
import java.util.List;

import javax.annotation.Resource;

import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.HTableInterface;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.filter.BinaryComparator;
import org.apache.hadoop.hbase.filter.CompareFilter;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.codehaus.jettison.json.JSONArray;
import org.codehaus.jettison.json.JSONObject;
import org.springframework.data.hadoop.hbase.HbaseTemplate;
import org.springframework.data.hadoop.hbase.RowMapper;
import org.springframework.data.hadoop.hbase.TableCallback;
import org.springframework.stereotype.Service;

import com.stoneframe.base.util.Pagination;

@Service(value="trackHbaseService")
public class TrackHbaseService {
	
	@Resource(name="hbaseTemplate")
	private HbaseTemplate hbaseTemplate;
	
	private String tableName="track";
	private String family="info";
	
	
	
	/**
	 * 查询列族下的列的数据
	 * */
	public List<String> listAll(){
		List<String> rows=hbaseTemplate.find(tableName, family, new RowMapper<String>(){

			public String mapRow(Result result, int rowNum) throws Exception {
				// TODO Auto-generated method stub
				return result.getRow().toString();
			}
			
		});
		return rows;
	}
	
	public List<String> listPage(Pagination page){
		Scan scan=new Scan();	//扫描器
		scan.setCaching(1000);	//缓存1000条数据,一次读取1000条
		RowFilter filter=new RowFilter(CompareFilter.CompareOp.EQUAL,
				new BinaryComparator("Jack".getBytes()));
		RowFilter filter1=new RowFilter(CompareFilter.CompareOp.EQUAL,
				new RegexStringComparator("J\\w+"));
		scan.setFilter(filter);
		
		PageFilter pageFilter=new PageFilter(page.getRows());
		scan.setFilter(pageFilter);
		scan.setStartRow(String.valueOf(page.getIndexofPage()).getBytes());
		
		List<String> rows=hbaseTemplate.find(tableName, scan, new RowMapper<String>(){

			public String mapRow(Result result, int rowNum) throws Exception {
				// TODO Auto-generated method stub
				return result.getRow().toString();
			}
			
		});
		return rows;
	}
	
	/**
	 * 查询指定行健的一列数据
	 * */
	public String get(String rowKey){
		String context=hbaseTemplate.get(tableName, rowKey, family, new RowMapper<String>(){

			public String mapRow(Result result, int rowNum) throws Exception {
				// TODO Auto-generated method stub
				return result.value().toString();
			}
			
		});
		return context;
	}
	
	/**
	 * 插入数据
	 * */
	public void save(final String rowKey,final JSONArray data){
		hbaseTemplate.execute(tableName, new TableCallback<Boolean>(){

			public Boolean doInTable(HTableInterface table) throws Throwable {
				// TODO Auto-generated method stub
				try{
					List<Put> puts=new ArrayList<Put>();
					for(int i=0;i<data.length();i++){
						JSONObject json=data.getJSONObject(i);
						Put put=new Put(rowKey.getBytes());
						put.add("info".getBytes(), "trackId".getBytes(), json.getString("trackId").getBytes());
						put.add("info".getBytes(), "terinalPN".getBytes(), json.getString("terinalPN").getBytes());
						put.add("info".getBytes(), "trackTime".getBytes(), json.getString("trackTime").getBytes());
						put.add("info".getBytes(), "trackLat".getBytes(), json.getString("trackLat").getBytes());
						put.add("info".getBytes(), "trackLon".getBytes(), json.getString("trackLon").getBytes());
						puts.add(put);
					}
					table.put(puts);
					return true;
				}catch(Exception e){
					e.printStackTrace();
				}
				return false;
			}
			
		});
	}
	
	/**
	 * 删除数据
	 * */
	public boolean del(final String rowKey){
		return hbaseTemplate.execute(tableName, new TableCallback<Boolean>(){

			public Boolean doInTable(HTableInterface htable) throws Throwable {
				// TODO Auto-generated method stub
				boolean flag=false;
				try{
					Delete delete=new Delete(rowKey.getBytes());
				}catch(Exception e){
					e.printStackTrace();
				}
				return flag;
			}
			
		});
	}
	

}
